package com.aaa.rabbitmq.four;

import com.aaa.rabbitmq.one.util.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认模式
 *  使用的时间 来比较那种确认方式最好的
 *  1，单个确认
 *  2，批量确认
 *  3，异步批量确认
 */
public class ConfirmMessage {

    //批量发消息的个数
    public static final int MESSAGE_COUNT= 1000;

    public static void main(String[] args) throws Exception {
        //1，单个确认    (同步确认，简单，吞吐量有限)
        //publishMessageIndividually();        //单个确认发布时间耗时：400ms
        //2，批量确认  （有合理的吞吐量，但无法确定是哪个消息丢失） 速度快
        //publishMessageBatch();                 //批量确认发布时间耗时：33ms
        //3，异步批量确认  （生产者发送一千条消息给信道（队列），消息类似于一个map 将消息序号作为key 每个消息信息作为value,将来broker（mq消息实体）收到消息会确认收到回调ackCallback，未确认收到回调nackCallback，并且是异步）
        //publishMessageAsync();                  //异步批量确认发布时间耗时：51ms
        //4, 异步批量确认完善 （最佳性能和资源使用，出现问题可以控制）
        publishMessageAsync01();                //异步批量确认发布时间耗时：34ms
    }

    //单个确认
    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String mesg = i + "";
            channel.basicPublish("",queueName,null,mesg.getBytes(StandardCharsets.UTF_8));
            //单个发送消息就马上确认
            boolean b = channel.waitForConfirms();
            if (b){
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println(
                "单个确认发布时间耗时："+(end-begin) +"ms"
        );
    }


    //批量发消息，批量发布确认
    public static void publishMessageBatch() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量确认消息大小
        int batchSize=100;
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String mesg = i + "";
            channel.basicPublish("",queueName,null,mesg.getBytes(StandardCharsets.UTF_8));
            //判断达到100条消息的时候 批量确认一次
            if (i%batchSize==0){
                channel.waitForConfirms();
                System.out.println("消息发送成功");
            }
        }


        //结束时间
        long end = System.currentTimeMillis();
        System.out.println(
                "批量确认发布时间耗时："+(end-begin) +"ms"
        );
    }

    //异步发布确认
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();



        /**
         * 消息确认成功回调函数
         * 1，消息的标识
         * 2，消息是否批量确认
         */
        ConfirmCallback ackCallback=(deliveryTag,multiple) ->{
            System.out.println("确认的消息编号"+deliveryTag);
        };
        /**
         * 消息确认失败回调函数
         * 1，消息的标识
         * 2，消息是否批量确认
         */
        ConfirmCallback nackCallback=(deliveryTag,multiple) ->{
            System.out.println("未确认的消息编号"+deliveryTag);
        };

        /**
         *         消息监听器 哪些消息成功哪些失败
         *         1,那些成功消息确认成功
         *         2,哪些消息确认失败
         */
        channel.addConfirmListener(ackCallback,nackCallback);//异步

        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String mesg = i + "";
            channel.basicPublish("",queueName,null,mesg.getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送成功");
        }


        //结束时间
        long end = System.currentTimeMillis();
        System.out.println(
                "异步批量确认发布时间耗时："+(end-begin) +"ms"
        );
    }


    /**
     * 异步发布确认
     * 问题 程序从上往下执行 当到达监听器时监听器未启动  当开始发送的时候监听器才启动  发送完毕监听器未执行完毕  涉及到两个线程的并发和交互 （发布线程和监听线程）
     * 解决方案把未确认的消息放到一个基于内存的能被发布线程访问的队列   ConcurrentLinkedQueue（并发链路队列） 这个队列在confirm callback与发布线程之间进行消息的传递
     *
     */

    public static void publishMessageAsync01() throws Exception{
        Channel channel = RabbitMQUtil.getChannel();
        //队列的声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表（Map） 适用于高并发的情况
         * 1,轻松的将序号与消息进行关联
         * 2，轻松的批量删除条目内容 只要给到序号（key）
         * 3，支持高并发（多线程）
         */
        ConcurrentSkipListMap<Long,String> outstandingConfirms = new ConcurrentSkipListMap<>();




        /**
         * 消息确认成功回调函数
         * 1，消息的标识
         * 2，消息是否批量确认
         */
        ConfirmCallback ackCallback=(deliveryTag,multiple) ->{
            //如果是批量确认
            if (multiple){
                //2，删除到已经确认的消息 剩下的就是确认的消息
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap = outstandingConfirms.headMap(deliveryTag);
                longStringConcurrentNavigableMap.clear();
            }else {
                outstandingConfirms.remove(deliveryTag);
            }

            System.out.println("确认的消息编号"+deliveryTag);
        };
        /**
         * 消息确认失败回调函数
         * 1，消息的标识
         * 2，消息是否批量确认
         */
        ConfirmCallback nackCallback=(deliveryTag,multiple) ->{
            //3 打印未确认的消息
            String s = outstandingConfirms.get(deliveryTag); //可以获取到未确认的消息
            System.out.println("未确认的消息是"+s+"-----"+"未确认的消息编号"+deliveryTag);
        };

        /**
         *         消息监听器 哪些消息成功哪些失败
         *         1,那些成功消息确认成功
         *         2,哪些消息确认失败
         */
        channel.addConfirmListener(ackCallback,nackCallback);//异步
        //开始时间
        long begin = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String mesg = i + "";
            channel.basicPublish("",queueName,null,mesg.getBytes(StandardCharsets.UTF_8));
            //1，此处记录所有要发送的消息，消息总和
            outstandingConfirms.put(channel.getNextPublishSeqNo(),mesg);

        }


        //结束时间
        long end = System.currentTimeMillis();
        System.out.println(
                "异步批量确认发布时间耗时："+(end-begin) +"ms"
        );
    }

}
